package indi.tutubrother.rocktmq.producer;

import indi.tutubrother.rocktmq.exception.RocketmqRuntimeException;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

/**
 * @author tutubrother
 * @description
 * @date 2020/5/5 15:46
 */
public abstract class BaseProducer {

    private static final Logger logger = LoggerFactory.getLogger(BaseProducer.class);

    protected String namesrvAddress;
    protected String topic;
    protected String producerGroup;
    protected DefaultMQProducer defaultMQProducer;

    protected BaseProducer(String topic, String producerGroup) {
        this.topic = topic;
        this.producerGroup = producerGroup;
    }

    /**
     * <p>
     * 初始化namesrvAddress，调用{@link BaseProducer#setNamesrvAddress(String)}方法
     * </p>
     *
     * @return void
     * @author tutubrother
     * @date 2020/5/5 15:49
     */
    protected abstract void init();

    protected void setNamesrvAddress(String namesrvAddress) {
        this.namesrvAddress = namesrvAddress;
    }

    public void start() {
        init();
        if (null == this.namesrvAddress) {
            throw RocketmqRuntimeException.build("请初始化namesrvAddress");
        }
        defaultMQProducer = new DefaultMQProducer(producerGroup);
        defaultMQProducer.setNamesrvAddr(namesrvAddress);
        try {
            defaultMQProducer.start();
            logger.info("producer(topic={}) starts success!", this.topic);
        } catch (MQClientException e) {
            logger.error("RocketMq starts failed:{}, topic:{}",
                    e.getErrorMessage(), this.topic);
        }
    }

}
